package com.bleeth.flow.core.common;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.ArrayUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import com.bleeth.flow.core.util.SparkUtil;
import com.jd.platform.async.executor.Async;
import com.jd.platform.async.wrapper.WorkerWrapper;
import lombok.Data;
import org.apache.log4j.Logger;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

/**
 * @author ：Bleeth
 * @date ：2021-08-06
 * @description：代表一个任务流程
 */

@Data
public class FlowTask {

    private Logger logger = Logger.getLogger(getClass());


    private String flowUid;

    private String flowName;

    private String flowCron;

    private JSONObject options;


    private Map<String, APlugin> flowPluginMap = new ConcurrentHashMap<>();

    private Map<String, WorkerWrapper<Dataset<Row>, Dataset<Row>>> wrapperMap = new ConcurrentHashMap<>();


    public static FlowTaskBuilder builder() {
        return new FlowTaskBuilder();
    }

    public void addPlugin(String name, APlugin plugin) {
        if (flowPluginMap == null) {
            flowPluginMap = new ConcurrentHashMap<>();
        }
        flowPluginMap.put(name, plugin);
    }

    public Collection<APlugin> getPlugins() {
        return flowPluginMap.values();
    }


    public void buildWrapper(boolean ok) throws Exception {
        List<APlugin> plugins = CollUtil.newArrayList(flowPluginMap.values());
        //对所有的plugins分析其子plugin
        int createIndex = 0;
        while (true) {
            for (int i = 0; i < plugins.size(); i++) {
                APlugin aPlugin = plugins.get(i);

                //已经创建完成
                if (wrapperMap.containsKey(aPlugin.getName())) {
                    continue;
                }

                List<String> nextList = aPlugin.getNextList();

                //没有下一步组件，可以创建ww
                if (CollUtil.isEmpty(nextList)) {
                    WorkerWrapper<Dataset<Row>, Dataset<Row>> ww = new WorkerWrapper.Builder<Dataset<Row>, Dataset<Row>>()
                            .worker(aPlugin)
                            .callback(aPlugin)
                            .id(aPlugin.getName())
                            .build();
                    wrapperMap.put(aPlugin.getName(), ww);
                    continue;
                }


                boolean createFlag = true;
                for (int nextIndex = 0; nextIndex < nextList.size(); nextIndex++) {
                    String nextName = nextList.get(nextIndex);

                    //设置from组件
                    APlugin nextPlugin = flowPluginMap.get(nextName);
                    List<String> fromList = nextPlugin.getFromList();
                    if (CollUtil.isEmpty(fromList)) {
                        fromList = CollUtil.newArrayList();
                        nextPlugin.setFromList(fromList);
                    }
                    if (!fromList.contains(aPlugin.getName())) {
                        fromList.add(aPlugin.getName());
                    }

                    if (!wrapperMap.containsKey(nextName)) {
                        createFlag = false;
                        break;
                    }
                }

                //下一步组件都创建了，可以创建ww
                if (createFlag) {
                    WorkerWrapper.Builder<Dataset<Row>, Dataset<Row>> builder = new WorkerWrapper.Builder<Dataset<Row>, Dataset<Row>>()
                            .worker(aPlugin)
                            .callback(aPlugin)
                            .id(aPlugin.getName());
                    List<WorkerWrapper> nextWW = nextList.stream().map(nextName -> {
                        WorkerWrapper<Dataset<Row>, Dataset<Row>> ww = wrapperMap.get(nextName);
                        return ww;
                    }).collect(Collectors.toList());
                    builder.next(ArrayUtil.toArray(nextWW, WorkerWrapper.class));
                    WorkerWrapper<Dataset<Row>, Dataset<Row>> ww = builder.build();
                    wrapperMap.put(aPlugin.getName(), ww);
                }
            }

            createIndex++;
            if (createIndex >= 100) {
                throw new Exception("配置出现问题，任务无法构建");
            }
            //是不是所有组件都创建完了，是的话就完成创建
            boolean buildFlag = true;
            for (APlugin plugin : plugins) {
                String name = plugin.getName();
                if (!wrapperMap.containsKey(name)) {
                    buildFlag = false;
                }
            }
            if (buildFlag) {
                break;
            }
        }
    }

    public void buildWrapper(int ok) throws Exception {
        List<APlugin> plugins = CollUtil.newArrayList(flowPluginMap.values());
        Map<String, WorkerWrapper.Builder> builderMap = MapUtil.newHashMap();

        //设置builder
        for (APlugin plugin : plugins) {
            WorkerWrapper.Builder<Dataset<Row>, Dataset<Row>> builder = new WorkerWrapper.Builder<Dataset<Row>, Dataset<Row>>()
                    .worker(plugin)
                    .callback(plugin)
                    .id(plugin.getName());
            builderMap.put(plugin.getName(), builder);
        }

        //设置from
        for (APlugin plugin : plugins) {
            List<String> nextList = plugin.getNextList();
            for (String nextName : nextList) {
                //设置from组件
                APlugin nextPlugin = flowPluginMap.get(nextName);
                List<String> fromList = nextPlugin.getFromList();
                if (CollUtil.isEmpty(fromList)) {
                    fromList = CollUtil.newArrayList();
                    nextPlugin.setFromList(fromList);
                }
                if (!fromList.contains(plugin.getName())) {
                    fromList.add(plugin.getName());
                }
            }
        }

        //设置
        for (int toIndex = 0, maxIndex = 100; toIndex < maxIndex; toIndex++) {
            for (APlugin plugin : plugins) {
                List<String> nextList = plugin.getNextList();
                boolean canBuildFlag = true;
                for (String nextName : nextList) {
                    if (wrapperMap.get(nextName) == null) {
                        canBuildFlag = false;
                        break;
                    }
                }
                if (canBuildFlag) {
                    WorkerWrapper.Builder builder = builderMap.get(plugin.getName());
                    if (CollUtil.isNotEmpty(nextList)) {
                        List<WorkerWrapper> nextWW = nextList.stream().map(nextName -> {
                            WorkerWrapper<Dataset<Row>, Dataset<Row>> ww = wrapperMap.get(nextName);
                            return ww;
                        }).collect(Collectors.toList());
                        builder.next(ArrayUtil.toArray(nextWW, WorkerWrapper.class));
                    }
                    WorkerWrapper<Dataset<Row>, Dataset<Row>> ww = builder.build();
                    wrapperMap.put(plugin.getName(), ww);
                }
            }
        }
    }


    public void buildWrapper() {
        List<APlugin> plugins = CollUtil.newArrayList(flowPluginMap.values());
        for (int i = 0; i < plugins.size(); i++) {
            APlugin aPlugin = plugins.get(i);
            recursionBuild(aPlugin);
        }
    }

    private void recursionBuild(APlugin plugin) {
        List<String> nextList = plugin.getNextList();

        if (CollUtil.isEmpty(nextList)) {
            WorkerWrapper<Dataset<Row>, Dataset<Row>> ww = new WorkerWrapper.Builder<Dataset<Row>, Dataset<Row>>()
                    .worker(plugin)
                    .callback(plugin)
                    .id(plugin.getName())
                    .build();
            wrapperMap.put(plugin.getName(), ww);
            return;
        } else {
            for (int i = 0; i < nextList.size(); i++) {
                String nextName = nextList.get(i);
                APlugin nextPlugin = flowPluginMap.get(nextName);

                //设置fromList
                List<String> fromList = nextPlugin.getFromList();
                if (CollUtil.isEmpty(fromList)) {
                    fromList = CollUtil.newArrayList();
                    nextPlugin.setFromList(fromList);
                }
                if (!fromList.contains(plugin.getName())) {
                    fromList.add(plugin.getName());
                }

                recursionBuild(nextPlugin);
                WorkerWrapper.Builder<Dataset<Row>, Dataset<Row>> builder = new WorkerWrapper.Builder<Dataset<Row>, Dataset<Row>>()
                        .worker(plugin)
                        .callback(plugin)
                        .id(plugin.getName());
                builder.next(wrapperMap.get(nextName), true);
                WorkerWrapper<Dataset<Row>, Dataset<Row>> ww = builder.build();
                wrapperMap.put(plugin.getName(), ww);
            }
        }
    }


    public static class FlowTaskBuilder {

        private String flowUid;

        private String flowName;

        private String flowCron;

        private JSONObject options;


        public FlowTaskBuilder flowUid(String flowUid) {
            this.flowUid = flowUid;
            return this;
        }

        public FlowTaskBuilder flowName(String flowName) {
            this.flowName = flowName;
            return this;
        }

        public FlowTaskBuilder flowCron(String flowCron) {
            this.flowCron = flowCron;
            return this;
        }

        public FlowTaskBuilder options(JSONObject options) {
            this.options = options;
            return this;
        }

        public FlowTask create() {
            FlowTask flowTask = new FlowTask();
            flowTask.setFlowName(flowName);
            flowTask.setFlowUid(flowUid);
            flowTask.setFlowCron(flowCron);
            return flowTask;
        }
    }



    public Object preview(String title, boolean b) {
        return null;
    }


    public void run() {
        boolean isCron = false;
        run(isCron);
    }

    public void run(boolean isCron) {
        TimeInterval timer = DateUtil.timer();

        List<WorkerWrapper> inputWrapperList = wrapperMap.entrySet().stream().filter(entry -> {
            String key = entry.getKey();
            APlugin aPlugin = flowPluginMap.get(key);
            return aPlugin.type.equalsIgnoreCase("INPUT");
        }).map(entry -> {
            return entry.getValue();
        }).collect(Collectors.toList());

        //Sparks设置
        if(options!=null){
            SparkSession spark = SparkUtil.getSparkInstance();
            for (Map.Entry<String, Object> entry : options.entrySet()) {
                spark.conf().set(entry.getKey(),Convert.toStr(entry.getValue()));
            }
        }

        logger.info(StrUtil.format("flow 开始执行：" + flowUid));
        try {
            Async.beginWork(Convert.toLong(Integer.MAX_VALUE), ThreadUtil.newExecutor(inputWrapperList.size()), inputWrapperList);
        } catch (ExecutionException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        Async.shutDown();
        long interval = timer.interval();//花费毫秒数
        logger.info(StrUtil.format("程序执行完成...，耗时{}min-{}sec-{}millis", interval / 1000 / 60, interval / 1000, interval % 1000));
    }


    public Map<String, APlugin> getFlowPluginMap() {
        return flowPluginMap;
    }


    public String toFlowString() {
        JSONObject object = new JSONObject();
        object.set("name", flowName);
        object.set("uuid", flowUid);
        object.set("cron", flowCron);

        JSONArray jobArray = new JSONArray();
        flowPluginMap.entrySet().stream().forEach(entry -> {
            APlugin aPlugin = entry.getValue();
            JSONObject jsonObject = aPlugin.toPluginString();
            jobArray.add(jsonObject);
        });
        object.set("job", jobArray);

        return object.toString();
    }
}
